package com.reload.reaactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;

import java.io.IOException;
import java.time.Duration;
import java.util.function.Consumer;

/**
 * @Date: 2024/4/25 上午11:23
 * @Version: 17
 * @Author: Reloading
 * @Description: 关于Flux和Mono
 */
public class FluxAndMono {
    public static void main(String[] args) {
        // Mono 是 0|1 个元素
        // Flux 是 N个元素
        FluxAndMono fluxAndMono = new FluxAndMono();
        // 测试Flux
        // fluxAndMono.flux();
        // 测试Mono
        System.out.println("=====测试Mono=====");
        // fluxAndMono.mono();
        // Flux事件感知
        System.out.println("=====Flux事件感知=====");
        fluxAndMono.eventAware();
        try {
            System.in.read();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 测试Flux
     */
    public void flux() {
        // 发布者发布数据
        Flux<Integer> just = Flux.just(1, 2, 3, 4, 5);
        // 流只要不消费就没用 -- 消费就是订阅
        Consumer<Integer> consumer = System.out::println;
        just.subscribe(e -> System.out.println("e1 = " + e));
        // 一个数据流可以有多个消费者
        just.subscribe(e -> System.out.println("e2 = " + e));
        // 对于每个消费者来说流都是一样的
        System.out.println("========================");
        // 每秒产生一个从0开始的递增数字
        Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));
        interval.subscribe(System.out::println);
    }

    /**
     * 测试Mono
     */
    public void mono() {
        Mono<Integer> just = Mono.just(1);
        just.subscribe(System.out::println);
    }

    /**
     * Flux事件感知
     * 响应式编程核心：看懂文档弹珠图
     * SignalType：
     *      SUBSCRIBE： 被订阅
     *      REQUEST：  请求了N个元素
     *      CANCEL： 流被取消
     *      ON_SUBSCRIBE：在订阅时候
     *      ON_NEXT： 在元素到达
     *      ON_ERROR： 在流错误
     *      ON_COMPLETE：在流正常完成时
     *      AFTER_TERMINATE：中断以后
     *      CURRENT_CONTEXT：当前上下文
     *      ON_CONTEXT：感知上下文
     * doOnXxx API触发时机
     *      1、doOnNext：每个数据（流的数据）到达的时候触发
     *      2、doOnEach：每个元素（流的数据和信号）到达的时候触发
     *      3、doOnRequest： 消费者请求流元素的时候
     *      4、doOnError：流发生错误
     *      5、doOnSubscribe: 流被订阅的时候
     *      6、doOnTerminate： 发送取消/异常信号中断了流
     *      7、doOnCancle： 流被取消
     *      8、doOnDiscard：流中元素被忽略的时候
     */
    public void eventAware() {
        Flux<Integer> flux = Flux.just(1, 2, 3,4,5,6,7)
                // 延迟发送
                .delayElements(Duration.ofSeconds(1))
                .doOnComplete(() -> System.out.println("流正常结束....."))
                .doOnCancel(() -> System.out.println("流被取消....."))
                // 这里的异常是处理本内的异常，不是订阅者的异常 对发布者的事件绑定
                .doOnError(throwable -> System.out.println("流异常结束" + throwable.getMessage()))
                .doOnNext(integer -> System.out.println("doOnNext = " + integer));
        // flux.subscribe(System.out::println);

        flux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("订阅者和发布者绑定" + subscription);
                // 背压
                request(1);
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("元素到达： " + value);
                if (value < 5) {
                    request(1);
                    if (value == 3) {
                        int i = 10 /0;
                    }
                }else {
                    cancel();
                }
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("数据流结束");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("数据流异常");
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("数据流被取消");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("结束信号： " + type);
            }
        });

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
